package com.bigdata.assignment.problem2;

import java.io.IOException;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

/**
 * WordCountCombiner
 * Combiner 的逻辑与 Reducer 完全一致：对同一个 key 在 map 端做局部聚合，
 * 以减少发送到 Reducer 的网络流量。
 *
 * 统计：
 * - COMBINER_INPUT: combiner 接收到的 input records（values 数量）
 * - COMBINER_OUTPUT: combiner 输出的 records（每个 key 输出一次）
 */
public class WordCountCombiner extends Reducer<Text, IntWritable, Text, IntWritable> {

    public static enum COUNTERS {
        COMBINER_INPUT,
        COMBINER_OUTPUT
    }

    private IntWritable result = new IntWritable();

    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context)
            throws IOException, InterruptedException {

        int localSum = 0;
        int valueCount = 0;
        for (IntWritable v : values) {
            localSum += v.get();
            valueCount++;
        }

        // Combiner 输入记录数：将本次 reduce 遍历的 value 数量计入 counter
        context.getCounter(COUNTERS.COMBINER_INPUT).increment(valueCount);

        result.set(localSum);
        context.write(key, result);

        // Combiner 输出记录数：每次对一个 key 输出一个记录
        context.getCounter(COUNTERS.COMBINER_OUTPUT).increment(1);
    }
}
